package com.atguigu.gmall.realtime.utils

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializeConfig
import org.apache.http.HttpHost
import org.elasticsearch.action.bulk.BulkRequest
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.client.{RequestOptions, RestClient, RestClientBuilder, RestHighLevelClient}
import org.elasticsearch.common.xcontent.XContentType

/**
 * @author caodan
 * @date 2022-09-13 16:46
 * @version 1.0
 */
object ESUtil {
  //声明es客户端
  var esClient: RestHighLevelClient = build()

  /**
   * 批量数据幂等写入
   * 通过指定id实现幂等
   */
  def bulkSaveIdempotent(sourceList: List[(String, AnyRef)], indexName: String): Unit = {
    if (sourceList != null && sourceList.nonEmpty) {
      // BulkRequest实际上就是由7多个单条IndexRequest的组合
      val bulkRequest = new BulkRequest()

      for ((docId, sourceObj) <- sourceList) {
        val indexRequest = new IndexRequest()
        indexRequest.index(indexName)
        val movieJsonStr: String = JSON.toJSONString(sourceObj, new SerializeConfig(true))
        indexRequest.source(movieJsonStr, XContentType.JSON)
        indexRequest.id(docId)
        bulkRequest.add(indexRequest)
      }

      esClient.bulk(bulkRequest, RequestOptions.DEFAULT)
    }

  }

  /**
   * 批量数据写入
   */
  def bulkSave(sourceList: List[AnyRef], indexName: String): Unit = {
    // BulkRequest实际上就是由多个单条IndexRequest的组合
    val bulkRequest = new BulkRequest()

    for (source <- sourceList) {
      val indexRequest = new IndexRequest()
      indexRequest.index(indexName)
      val movieJsonStr: String = JSON.toJSONString(source, new SerializeConfig(true))
      indexRequest.source(movieJsonStr, XContentType.JSON)
      bulkRequest.add(indexRequest)
    }

    esClient.bulk(bulkRequest, RequestOptions.DEFAULT)
  }

  /**
   * 单条数据幂等写入
   * 通过指定id实现幂等
   */
  def saveIdempotent(source: (String, AnyRef), indexName: String): Unit = {
    val indexRequest = new IndexRequest()
    indexRequest.index(indexName)
    val movieJsonStr: String = JSON.toJSONString(source._2, new SerializeConfig(true))
    indexRequest.source(movieJsonStr, XContentType.JSON)
    indexRequest.id(source._1)
    esClient.index(indexRequest, RequestOptions.DEFAULT)
  }

  /**
   * 单条数据写入
   */
  def save(source: AnyRef, indexName: String): Unit = {
    val indexRequest = new IndexRequest()
    indexRequest.index(indexName)
    val movieJsonStr: String = JSON.toJSONString(source, new SerializeConfig(true))
    indexRequest.source(movieJsonStr, XContentType.JSON)
    esClient.index(indexRequest, RequestOptions.DEFAULT)
  }

  /**
   * 销毁
   */
  def destory(): Unit = {
    esClient.close()
    esClient = null
  }

  /**
   * 创建es客户端对象
   */
  def build(): RestHighLevelClient = {
    val builder: RestClientBuilder = RestClient.builder(new HttpHost("hadoop102",9200))
    val esClient = new RestHighLevelClient(builder)
    esClient
  }

  /**
   * 获取esclient
   */
  def getClient: RestHighLevelClient = {
    esClient
  }

}
